package com.adam.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.filecache.DistributedCache;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.elasticsearch.hadoop.mr.EsOutputFormat;
import java.io.IOException;

public class HdfsToES {
    public static class MyMapper extends Mapper<Object, Text, NullWritable, Text> {
        //Mapper的四个泛型参数分别为KEYIN, VALUEIN, KEYOUT, VALUEOUT
        public void map(Object key, Text value, Mapper<Object, Text, NullWritable, Text>.Context context)
                throws IOException, InterruptedException {
            byte[] line = value.toString().trim().getBytes();
            Text blog = new Text(line);
            context.write(NullWritable.get(), blog);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        conf.setBoolean("mapred.map.tasks.speculative.execution", false);  //关闭Map阶段执行推测
        conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);
        conf.set("es.nodes", "192.168.0.112:9200");
        conf.set("es.net.http.auth.user", "elastic");
        conf.set("es.net.http.auth.pass", "elastic");
        conf.set("es.resource", "test_blog"); //索引到ES的索引名和类型名
        conf.set("es.mapping.id", "id"); //指定id字段，不指定会随机生成
        conf.set("es.input.json", "yes"); //指定输入文件类型为json
        //DistributedCache.addFileToClassPath(new Path("hdfs://192.168.0.112:9000//library/elasticsearch-hadoop-mr-5.4.0.jar"), conf);  //添加到job执行容器classpath

        Job job = Job.getInstance(conf, "hadoop es write test");
        job.setJarByClass(HdfsToES.class);
        job.setMapperClass(HdfsToES.MyMapper.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(EsOutputFormat.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.0.112:9000//test_hdfs_to_es/input/blog.json"));
        job.waitForCompletion(true);
    }
}